package com.atguigu.day05;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用union合流之后用coProcessFunction聚合
 */
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Tuple2<String, Integer>> stream1 = env.fromElements(
                Tuple2.of("a", 1),
                Tuple2.of("c", 1),
                Tuple2.of("b", 1)
        );

        DataStreamSource<Tuple2<String, Integer>> stream2 = env.fromElements(
                Tuple2.of("a", 2),
                Tuple2.of("c", 3),
                Tuple2.of("c", 1),
                Tuple2.of("c", 8),
                Tuple2.of("b", 4),
                Tuple2.of("b", 7)
        );
        stream1
                .keyBy(x -> x.f0)
                .broadcast()
                .connect(stream2.keyBy(x -> x.f0))
                .process(new CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String>() {
                    private ListState<Tuple2<String, Integer>> list1;
                    private ListState<Tuple2<String, Integer>> list2;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        list2 = getRuntimeContext().getListState(new ListStateDescriptor<Tuple2<String, Integer>>(
                                "list2", Types.TUPLE(Types.STRING, Types.INT)
                        ));
                        list1 = getRuntimeContext().getListState(new ListStateDescriptor<Tuple2<String, Integer>>(
                                "list1", Types.TUPLE(Types.STRING, Types.INT)
                        ));
                    }

                    @Override
                    public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
                        list1.add(value);
                        list2.get().forEach(s2 -> {
                            out.collect(value + "=>" + s2);
                        });
                    }

                    @Override
                    public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
                        list2.add(value);
                        list1.get().forEach(s1 -> {
                            out.collect(value + "=>" + s1);
                        });
                    }
                })
                .print();

        env.execute();
    }
}
